package com.cskaoyan.rocketmq.transaction;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.Charset;

/**
 * 创建日期: 2023/01/18 15:17
 *
 * @author ciggar
 */
public class TransactionProducer {

    // 事务型消息

    public static void main(String[] args) throws MQClientException {

        // 1. 创建一个Producer
        TransactionMQProducer transactionProducer = new TransactionMQProducer("transaction_mq_producer");

        // 2. 设置注册中心
        transactionProducer.setNamesrvAddr("localhost:9876");

        // 3. 设置事务监听器 (对比传统消息，增加的一步)
        transactionProducer.setTransactionListener(new TransactionListener() {

            // 执行本地事务
            /**
             * @param msg  消息对象
             * @param arg  发送事务型消息的时候，传递的arg参数   sendMessageInTransaction(message, arg)
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                System.out.println("executeLocalTransaction ... 执行本地事务, transactionId:" + msg.getTransactionId());

                // 返回本地事务执行成功
                // return LocalTransactionState.COMMIT_MESSAGE;

                // 返回本地事务执行失败
                // return LocalTransactionState.ROLLBACK_MESSAGE;

                // 返回UNKONW，那么就表示 RocketMQ不知道本地事务是否执行成功，就会调用回查的方法
                return LocalTransactionState.UNKNOW;

            }


            // 检查本地事务
            // 回查本地事务的执行状态
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("checkLocalTransaction... 检查本地事务的执行状态, transactionId:" + msg.getTransactionId());

                // 检查到本地事务执行成功
                // return LocalTransactionState.COMMIT_MESSAGE;

                // 检查到本地事务执行失败
                // return LocalTransactionState.ROLLBACK_MESSAGE;

                // 检查本地事务，不知道本地事务执行成功与否
                return LocalTransactionState.UNKNOW;
            }
        });


        // 4. 启动
        transactionProducer.start();

        // 5. 构建消息对象
        String content = "hello rocket mq transaction message";

        String topic = "transaction_test_topic";

        Message message = new Message("transaction_test_topic", content.getBytes(Charset.forName("utf-8")));

        // 6. 发送消息
        // 发送事务消息的时候，每一次事务消息，都有自己独一无二的事务Id
        // String transactionId = message.getTransactionId();
        TransactionSendResult sendResult = transactionProducer.sendMessageInTransaction(message, null);

        // 7. 打印消息的发送结果
        System.out.println(sendResult);

    }
}
